package com.rabbitmq.rabbitmqextensions.config;

import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

@Configuration
public class RabbitTemplateConfig {

    /**
     *
     * @param connectionFactory 会根据配置文件中的 rabbitmq 配置自动填充参数
     * @return
     */
    @Bean("normalRabbitTemplate")
    public RabbitTemplate normalRabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    /**
     * 确认模式使用
     * 使用 AtomicBoolean, 在消息被退回时打印不同的日志
     * @param connectionFactory
     * @return
     */
    @Bean("confirmRabbitTemplate1")
    public RabbitTemplate confirmRabbitTemplate1(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        // 判断消息是否发送成功
        //局部变量不能在匿名内部类修改
//        boolean flag = false;

        // 在消息被退回时打印不同的日志
        AtomicBoolean flag = new AtomicBoolean(true);

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            /**
             *
             * @param correlationData correlation data for the callback.
             * @param ack true: 消息到达交换机; false: 消息没有到达交换机
             * @param cause An optional cause, for nack, when available, otherwise null.
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                String messageId = null == correlationData ? null : correlationData.getId();
                if (!flag.get()) {
                    System.out.println("消息路由失败, 消息 id: " + messageId);
                } else if (ack) {
                    System.out.println("接收到消息, 消息 id: " + messageId);
                } else {
                    System.out.println("未接收到消息, 消息 id: " + messageId + ", cause: " + cause);
                }
            }
        });

        /**
         * 若消息没有到达队列，就退回
         */
        rabbitTemplate.setMandatory(true); //启用强制路由检查
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                flag.set(false);
                System.out.println("消息退退回: " + returned);
            }
        });

        return rabbitTemplate;
    }


    /**
     * 使用 ConcurrentHashMap 来判断消息是否被退回，并打印相应的消息
     * @param connectionFactory
     * @return
     */
    @Bean("confirmRabbitTemplate2")
    public RabbitTemplate confirmRabbitTemplate2(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        // 判断消息是否发送成功
        //局部变量不能在匿名内部类修改
//        boolean flag = false;

        // 在消息被退回时打印不同的日志
        //key 为消息 id，value 为该条消息是否被退回
        ConcurrentHashMap<String, Boolean> map = new ConcurrentHashMap<>();

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            /**
             *
             * @param correlationData correlation data for the callback.
             * @param ack true: 消息到达交换机; false: 消息没有到达交换机
             * @param cause An optional cause, for nack, when available, otherwise null.
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                String messageId = null == correlationData ? null : correlationData.getId();
                if (!map.getOrDefault(messageId, true)) {
                    System.out.println("消息路由失败, 消息 id: " + messageId);
                } else if (ack) {
                    System.out.println("接收到消息, 消息 id: " + messageId);
                } else {
                    System.out.println("未接收到消息, 消息 id: " + messageId + ", cause: " + cause);
                }
            }
        });

        /**
         * 若消息没有到达队列，就退回
         */
        rabbitTemplate.setMandatory(true); //启用强制路由检查
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
//                System.out.println(returned.getMessage().getMessageProperties().getMessageId());
//                System.out.println(returned.getMessage().getMessageProperties().getHeaders());

                // 从 headers 中获取 messageId
                // headers={spring_returned_message_correlation=5a53f71c-575c-402d-8b91-4a88599320d2}
                String headers = returned.getMessage().getMessageProperties().getHeaders().toString();
                int index1 = headers.indexOf('=');
                int index2 = headers.indexOf('}');
                String messageId = headers.substring(index1, index2);
                map.put(messageId, false);
                System.out.println("消息退退回: " + returned);
            }
        });

        return rabbitTemplate;
    }

    /**
     * 开启事务
     * @param connectionFactory
     * @return
     */
    @Bean("transRabbitTemplate")
    public RabbitTemplate transRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        rabbitTemplate.setChannelTransacted(true);

        return rabbitTemplate;
    }
}
